#include "pipe.h"
#include <memory>
#include "ecode.h"

#include <windows.h>

using namespace kog;

pipe::pipe(const std::string& pipe_name, port_type ptype)
: _pipe_name(pipe_name)
, _ptype(ptype)
, _hpipe(INVALID_HANDLE_VALUE)
, _isruning(false)
, _send_sem(NULL)
, _recv_sem(NULL)
, _hsend_thread(NULL)
, _hrecv_thread(NULL)
{
	_send_sem.create_semaphore(0, MAX_QUEUE_SIZE);
	_recv_sem.create_semaphore(0, MAX_QUEUE_SIZE);
	_send_guard.create_mutex(false);
	_recv_guard.create_mutex(false);
	
	// create pipe
	if (ptype == server_port)
	{
		HANDLE hPipe = CreateNamedPipeA(pipe_name.c_str(), PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
			PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE, 4, 0, 0, 1000, NULL);
		if (hPipe != INVALID_HANDLE_VALUE)
		{	
			_hpipe.reset(hPipe);
			// log create done
		}
		else
		{
			// log create failed!
		}
	}
}

pipe::~pipe(void)
{
	_cleanup();
}

bool pipe::is_valid() const
{
	return _hpipe.is_valid();
}

int pipe::waitclient(int timeout)
{
	int check_result = _check(server_port, true);
	if (check_result != 0) return check_result;

	int rcode = ConnectNamedPipe(_hpipe.get(), NULL);
	if (rcode != TRUE) 
	{
		printf("wait client failed! %d\n", GetLastError());
		// log message GetLastError();
		return -1;
	}
	
	// log string
	// start thread
	_isruning = true;
	_start_threads();
	return 0;
}

int pipe::connectserver(int timeout)
{
	int check_result = _check(client_port, false);
	if (check_result != 0) return check_result;
	
	if (WaitNamedPipeA(_pipe_name.c_str(), timeout) != TRUE)
	{
		DWORD ecode = GetLastError();
		if (ERROR_SEM_TIMEOUT == ecode)
		{
			printf("\n");
			return KOG_ERROR_HPIPE_TIMEOUT_CONNECT;
		}

		printf("connect server failed!\n");
		return KOG_ERROR_HPIPE_FAILED_CONNECT;
	}

	HANDLE hFile = CreateFileA(_pipe_name.c_str(), GENERIC_READ | GENERIC_WRITE, 
			FILE_SHARE_READ | FILE_SHARE_WRITE, NULL, OPEN_EXISTING,
			FILE_FLAG_OVERLAPPED, NULL);
	if (hFile == INVALID_HANDLE_VALUE) 
	{
		// log message
		printf("create file failed!\n");
		return KOG_ERROR_HPIPE_FAILED_CREATEFILE;
	}

	_hpipe.reset(hFile);

	// start thread here
	_isruning = true;
	_start_threads();
	return 0;
}

int pipe::_check(port_type ptype, bool chk_handle)
{
	if (_isruning) 
	{
		printf("is running.\n");
		return KOG_ERROR_ENTRY_AGAIN;
	}
	if (chk_handle && !is_valid()) 
	{
		printf("handle is invalidate .\n");
		return KOG_ERROR_HPIPE_INVALIDATE;
	}
	else if (_ptype != ptype)
	{
		printf("type conflict.\n");
		return KOG_ERROR_HPIPE_DTYPE_CONFLICT;
	}
	return 0;
}

void pipe::_cleanup()
{
	_isruning = false;

	if (_hpipe.is_valid())
	{
		_hpipe.close();
	}

	// wait thread stop here
	if (_hsend_thread.is_valid())
	{
		while (_hsend_thread.wait(500) == WAIT_TIMEOUT);
		printf("send thread stoped!\n");
	}

	if (_hrecv_thread.is_valid())
	{
		while (_hrecv_thread.wait(500) == WAIT_TIMEOUT);
		printf("recv thread stoped!\n");
	}

	// clean queue
	_send_guard.wait();
	while (!_tosend.empty())
	{
		message_t* pmeta = _tosend.front();
		_tosend.pop();
		delete [](char*)pmeta;
		printf("delete push data.\n");
	}
	_send_guard.release();

	_recv_guard.wait();
	while (!_recved.empty())
	{
		message_t meta = _recved.front();
		_recved.pop();
		delete []meta.data.ptr;
		printf("delete recv data.\n");
	}
	_recv_guard.release();

	// let someone warkup
	_recv_sem.release();

	_send_guard.close();
	_recv_guard.close();
	_send_sem.close();
	_recv_sem.close();
	_hsend_thread.close();
	_hrecv_thread.close();
}

void pipe::_start_threads()
{
	//std::bind1st
	_hsend_thread.create_thread(unary2zero_ref(std::mem_fun_ref(&pipe::_send_thread), *this));

	_hrecv_thread.create_thread(unary2zero_ref(std::mem_fun_ref(&pipe::_recv_thread), *this));

	// some error, stop threads
	if (_hsend_thread.get() == NULL || _hrecv_thread.get() == NULL)
	{
		_isruning = false;
	}
	Sleep(10);
}

unsigned long pipe::_waitfor(winobject hobject)
{
	int timeout = 500;
	while (_isruning)
	{
		int rcode = hobject.wait(timeout);
		if (WAIT_TIMEOUT == rcode)
		{
		}
		else 
		{
			return rcode;
		}
	}
	return WAIT_TIMEOUT;
}

void pipe::_send_thread()
{
	printf("send thread is ok.\n");
	winobject revent;
	revent.create_event(false, true);
	OVERLAPPED overlap;
	ZeroMemory(&overlap, sizeof(overlap));
	overlap.hEvent = revent.get();
	DWORD timeout = 500;
	while (_isruning)
	{
		message_t* pmeta = NULL;

		if(_send_sem.wait(timeout) != WAIT_OBJECT_0)
		{
			continue;
		}

		_send_guard.wait(/*INFINITE*/);
		if (!_tosend.empty())
		{
			pmeta = _tosend.front();
			_tosend.pop();
		}
		_send_guard.release();

		if (pmeta == NULL) continue;
		std::auto_ptr<char> tmp((char*)pmeta);

		DWORD written = 0;
		if (!WriteFile(_hpipe.get(), pmeta, pmeta->len, &written, &overlap) && GetLastError() != ERROR_IO_PENDING)
		{
			printf("call write failed! %d\n", GetLastError());
			printf("delete push data.\n");
			//TRACE1("call write failed! %d\n", GetLastError());
			//delete [](char*)pmeta;
			return;
		}

		DWORD rcode = _waitfor(revent);
		if (rcode == WAIT_OBJECT_0)
		{
			// log
		}

		//delete [](char*)pmeta;
		
		// delete data
		printf("delete push data.\n");
	}
}

void pipe::_recv_thread()
{
	printf("recv thread is ok.\n");
	winobject revent;
	revent.create_event(false, true);
	OVERLAPPED overlap;
	ZeroMemory(&overlap, sizeof(overlap));
	overlap.hEvent = revent.get();
	DWORD timeout = 500;

	while (_isruning)
	{
		message_t meta;
		memset(&meta, 0, sizeof(meta));

		DWORD read = 0;
		// waiting new data
		if (!::ReadFile(_hpipe.get(), &meta, sizeof(meta), &read, &overlap))
		{
			DWORD errorCode = GetLastError();
			if (ERROR_MORE_DATA != errorCode && ERROR_IO_PENDING != errorCode)
			{
				printf("call read failed! %d\n", errorCode);
				//system("echo call read failed! %d >> tmp.txt");
				return;
			}
		}

		DWORD rcode = _waitfor(revent);
		if (rcode != WAIT_OBJECT_0)
		{
			// log error
			return ;
		}

		if (meta.len < sizeof(meta) || meta.len > MAX_META_DATA)
		{
			// into unrecovered error.
			// logging here
			return ;
		}
		ulong leftsize = meta.len - sizeof(meta);
		ulong datasize = leftsize + sizeof(meta.data);
		std::auto_ptr<char> ptmp(new char[datasize]);
		memcpy(ptmp.get(), meta.data.quick, sizeof(meta.data));

		char* ptr = ptmp.get() + sizeof(meta.data);
		if (!::ReadFile(_hpipe.get(), ptr, leftsize, &read, &overlap))
		{
			DWORD errorCode = GetLastError();
			if (ERROR_MORE_DATA != errorCode && ERROR_IO_PENDING != errorCode)
			{
				printf("call read more failed! %d\n", errorCode);
				return;
			}
		}

		rcode = _waitfor(revent);
		if (rcode != WAIT_OBJECT_0)
		{
			// log error
			return;
		}

		meta.data.ptr = ptmp.get();
		printf("new recv data.\n");

		_recv_guard.wait();
		_recved.push(meta);
		_recv_guard.release();

		ptmp.release();

		// raise event
		_recv_sem.release();
	}

}

int pipe::send(int fid, int tid, int dtype, const msgpeice_t* ps, int n)
{
	if (!is_valid()) 
	{
		return KOG_ERROR_HPIPE_INVALIDATE;
	}
	else if (!_isruning) 
	{
		return KOG_ERROR_HPIPE_NOTRUNNING;
	}

	message_t* meta = NULL;
	_combine(ps, n, meta);
	meta->from = fid;
	meta->to = tid;
	meta->dtype = dtype;

	_send_guard.wait();
	_tosend.push(meta);
	_send_guard.release();
	// raise event
	_send_sem.release();
	return 0;
}

void pipe::_combine(const msgpeice_t* ps, int n, message_t*& result)
{
	message_t pmt;
	int Len = sizeof(message_t) - sizeof(pmt.data);
	for (int i = 0; i < n; ++ i)
	{
		Len += ps[i].len;
	}
	Len = Len < sizeof(message_t) ? sizeof(message_t) : Len;

	std::auto_ptr<char> tmp(new char[Len]);
	printf("new push data.\n");
	message_t* meta = (message_t*)tmp.get();
	meta->len = Len;
	char* p = (char*)meta->data.quick;
	for (int i = 0; i < n; ++ i)
	{
		memcpy(p, ps[i].ptr, ps[i].len);
		p += ps[i].len;
	}

	result = (message_t*)tmp.release();
}

int pipe::recv(int& fid, int& tid, int& dtype, msgpeice_t& msg, int timeout)
{
	if (!is_valid()) 
	{
		return KOG_ERROR_HPIPE_INVALIDATE;
	}
	else if (!_isruning) 
	{
		return KOG_ERROR_HPIPE_NOTRUNNING;
	}

	int r = _recv_sem.wait(timeout);
	if(r == WAIT_TIMEOUT)
	{
		return KOG_ERROR_HPIPE_TIMEOUT_RECV;
	}
	else if(r != WAIT_OBJECT_0)
	{
		return KOG_ERROR_UNKNOWN;
	}

	_recv_guard.wait();
	if (_recved.empty())
	{
		_recv_guard.release();
		return KOG_ERROR_HPIPE_QUEUE_EMPTY;
	}
	message_t meta = _recved.front();
	_recved.pop();
	_recv_guard.release();

	msg.ptr = meta.data.ptr;
	msg.len = meta.len - sizeof(meta) + sizeof(meta.data);
	fid = meta.from;
	tid = meta.to;
	dtype = meta.dtype;
	return 0;
}
